package com.wpml.netty.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.Channel;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Description
 * @Date 2021-07-16
 * @Author wangpei
 **/
public class TaskThreadGroup {

    TaskThread[] taskThreads;
    ServerSocketChannel server;
    AtomicInteger xid = new AtomicInteger(0);

    TaskThreadGroup workerGroup = this;

    public void setWorkerGroup(TaskThreadGroup workerGroup) {
        this.workerGroup = workerGroup;
    }


    TaskThreadGroup(int threadNum) {
        taskThreads = new TaskThread[threadNum];

        for (int i = 0; i < threadNum; i++) {
            taskThreads[i] = new TaskThread();

            new Thread(taskThreads[i]).start();
        }
    }


    public void bind(int port) {
        try {
            server = ServerSocketChannel.open();
            server.configureBlocking(false);
            server.bind(new InetSocketAddress(port));

            // 选择一个多路复用器进行注册
            chooseSelector(server);

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void chooseSelector(Channel c) {
        try {
            if (c instanceof ServerSocketChannel) {
                TaskThread next = next();
                next.lbq.put(c);
                // 设置工作线程组
                next.setWorkerGroup(workerGroup);
                // 唤醒
                next.selector.wakeup();
            } else if (c instanceof SocketChannel) {
                TaskThread taskThread = nextWorker();
                taskThread.lbq.put(c);
//                taskThread.setWorkerGroup(this);
                taskThread.selector.wakeup();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public TaskThread nextWorker() {
        int count = xid.getAndIncrement();
        int index = count % workerGroup.taskThreads.length;
        return workerGroup.taskThreads[index];
    }

    private TaskThread next() {
        int count = xid.getAndIncrement();
        int index = count % taskThreads.length;

        return taskThreads[index];
    }
}
